<?php
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
    class Sender{
        const HOST = 'localhost';
        const PORT = 5672;
        const USER = 'guest';
        const PASS = 'guest';
        const VHOST = 'site';

        /**
         * 发送队列消息
         * @param $exchange 交换器
         * @param $queue 队列
         * @param $message 消息
         * @param string $exchangeType 交换器类型 默认为direct
         * @return 消息入列状态
         */
        public static function sendMessage($exchange,$queue,$message,$exchangeType='direct',$key=''){
            require_once('autoload.php');
            $msgStatus = null;
            $conn = new AMQPStreamConnection(self::HOST,self::PORT,self::USER,self::PASS,self::VHOST);
            $chan = $conn->channel();
            //发送方确认模式
            $chan->confirm_select();
            $chan->exchange_declare($exchange,$exchangeType,false,true,false);
            $chan->queue_declare($queue,false,true,false,false);
            $chan->queue_bind($queue,$exchange,$key);
            $chan->set_ack_handler(function(AMQPMessage $msg)use(&$msgStatus){
                $result = true;
            });
            $chan->set_nack_handler(function(AMQPMessage $msg)use(&$msgStatus){
                $result = false;
            });
            $message = new AMQPMessage($message);
            $chan->basic_publish($message,$exchange,$key);
            //等待入列状态
            $chan->wait_for_pending_acks();
            $chan->close();
            $conn->close();
            return $msgStatus;
        }

        /**
         * 发送延时队列消息
         * @param $exchange 交换器
         * @param $queue 队列
         * @param $message 消息
         * @param $delay 延迟时间，单位毫秒
         * @param string $deadLetterExchange 死信交换器
         * @return null 消息入列状态
         */
       public static function sendDelayedMessage($exchange,$queue,$message,$delay,$deadLetterExchange='deadline_letter_exchange'){
            require('./autoload.php');
            $conn = new AMQPStreamConnection(self::HOST,self::PORT,self::USER,self::PASS,self::VHOST);
            $etable = new AMQPTable(array('x-delay-type'=>'direct'));
            $qtable = new AMQPTable(array('x-dead-letter-exchange'=>$exchange));
            $mtable = new AMQPTable(array('x-delay'=>$delay,'delivery_tag'=>2));
            $chan = $conn->channel();
           //发送方确认模式
            $chan->confirm_select();
            $msgStatus = null;
            $chan->exchange_declare($exchange,'x-delayed-message',false,true,false,false,false,$etable);
            $chan->queue_declare($queue,false,true,false,false,false,$qtable);
            $chan->queue_bind($queue,$exchange);
            $message = new AMQPMessage($message);
            $message->set('application_headers',$mtable);
            $chan->basic_ack(function(AMQPMessage $msg)use(&$msgStatus){
                $msgStatus = true;
            });
            $chan->basic_nack(function(AMQPMessage $msg)use(&$msgStatus){
                $msgStatus = false;
            });
            $chan->basic_publish($message,$exchange);
            //等待入列状态
            $chan->wait_for_pending_acks();
            $chan->close();
            $conn->close();
            return $msgStatus;
       }
    }